package cn.liuwenlin.mq;

import cn.liuwenlin.mq.listener.MessageListener;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;

/**
 * activemq数据接受
 * @author liuwenlin
 */
public class QueueReceiver2 {

    volatile static boolean sentinel = false;

    static Thread forkThread;

    public static void main(String[] args) {

        final Thread mainThread = Thread.currentThread();

        //测试中断
        forkThread = new Thread(new Runnable() {
            @Override
            public void run() {
                Scanner scanner = new Scanner(System.in);
                String inputString;
                while((inputString = scanner.next()) != null){
                    if("stop".equals(inputString) || "quit".equals(inputString)){
                        QueueReceiver2.stop(mainThread);
                        break;
                    }
                }
            }
        });

        forkThread.start();
        System.out.println("Thread listener thread started...");

        //1.建立连接工厂
//        ActiveMQConnectionFactory activeMQConnectionFactory
//                = new ActiveMQConnectionFactory(
//                ActiveMQConnectionFactory.DEFAULT_USER,
//                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
//                "tcp://master.tony.com:61616");
        ActiveMQConnectionFactory activeMQConnectionFactory
                = new ActiveMQConnectionFactory(
                ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                "nio://hadoop.master.com:5671");
        activeMQConnectionFactory.setTrustAllPackages(true);
        Connection connection = null;
        try {
            //2.获取一个向Active MQ的连接
            connection = activeMQConnectionFactory.createConnection();

            //3.获取session
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            //4.找目的地，获取destination，消费端，也会从这个目的地取消息
//            Destination mq = session.createQueue("activemq");
            Destination mq = session.createQueue("DLQ.activemq");

            //5.消息消费者
            MessageConsumer consumer = session.createConsumer(mq);
//            MessageConsumer consumer = session.createConsumer(mq, "ttl > 200");
//            MessageConsumer consumer = session.createConsumer(mq, "type=1");

            connection.start();
            System.out.println("Receiver has been started...");
            //6.接受消息，原始写法看起来不优雅，替换成监听器模式

//            while(!Thread.currentThread().isInterrupted()){
//                //这里是同步阻塞的
//                TextMessage message = (TextMessage)consumer.receive();
//                System.out.println("receive msg:" + message.getText());
//                TimeUnit.MILLISECONDS.sleep(100);
//            }

            consumer.setMessageListener(new MessageListener());

            while (!Thread.currentThread().isInterrupted()){
                TimeUnit.MILLISECONDS.sleep(10);
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {

            //6.关闭连接
            if(connection != null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }

            System.out.println("exit...");
        }

    }

    public static void stop(Thread thread){
        sentinel = true;
        thread.interrupt();
    }

}
